package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * job trigger thread pool helper
 *
 * @author xuxueli 2018-07-03 21:08:07
 */
public class JobTriggerPoolHelper {

  private static Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class);

  // ---------------------- trigger pool ----------------------

  // fast/slow thread pool
  private ThreadPoolExecutor fastTriggerPool = null;
  private ThreadPoolExecutor slowTriggerPool = null;

  public void start() {
    fastTriggerPool = new ThreadPoolExecutor(
        10,
        XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
        60L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(1000),
        new ThreadFactory() {
          @Override
          public Thread newThread(Runnable r) {
            return new Thread(r,
                "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode());
          }
        });

    slowTriggerPool = new ThreadPoolExecutor(
        10,
        XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
        60L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(2000),
        new ThreadFactory() {
          @Override
          public Thread newThread(Runnable r) {
            return new Thread(r,
                "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode());
          }
        });
  }


  public void stop() {
    // triggerPool.shutdown();
    fastTriggerPool.shutdownNow();
    slowTriggerPool.shutdownNow();
    logger.info(">>>>>>>>> xxl-job trigger thread pool shutdown success.");
  }


  // job timeout count
  private volatile long minTim = System.currentTimeMillis() / 60000;     // ms > min
  private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();


  /**
   * add trigger
   */
  public void addTrigger(final int jobId,
      final TriggerTypeEnum triggerType,
      final int failRetryCount,
      final String executorShardingParam,
      final String executorParam,
      final String addressList) {

    // choose thread pool
    ThreadPoolExecutor triggerPool_ = fastTriggerPool;
    AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
    if (jobTimeoutCount != null
        && jobTimeoutCount.get() > 10) {      // job-timeout 10 times in 1 min
      triggerPool_ = slowTriggerPool;
    }

    // trigger
    triggerPool_.execute(new Runnable() {
      @Override
      public void run() {

        long start = System.currentTimeMillis();

        try {
          // do trigger
          XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam,
              executorParam, addressList);
        } catch (Exception e) {
          logger.error(e.getMessage(), e);
        } finally {

          // check timeout-count-map
          long minTim_now = System.currentTimeMillis() / 60000;
          if (minTim != minTim_now) {
            minTim = minTim_now;
            jobTimeoutCountMap.clear();
          }

          // incr timeout-count-map
          long cost = System.currentTimeMillis() - start;
          if (cost > 500) {       // ob-timeout threshold 500ms
            AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId,
                new AtomicInteger(1));
            if (timeoutCount != null) {
              timeoutCount.incrementAndGet();
            }
          }

        }

      }
    });
  }

  // ---------------------- helper ----------------------

  private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper();

  public static void toStart() {
    helper.start();
  }

  public static void toStop() {
    helper.stop();
  }

  /**
   * @param jobId
   * @param triggerType
   * @param failRetryCount        >=0: use this param <0: use param from job info config
   * @param executorShardingParam
   * @param executorParam         null: use job param not null: cover job param
   */
  public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount,
      String executorShardingParam, String executorParam, String addressList) {
    helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam,
        addressList);
  }

}
